package com.nami.dispatcher.infra.middleware.mq;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import com.nami.dispatcher.domain.middleware.mq.DomainEventPublisher;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import java.io.UnsupportedEncodingException;

import static com.nami.gate.external.basic.constants.CommonConst.COLON;


@Slf4j
@RequiredArgsConstructor
@Service
public class DefaultDownMsgPublishImpl implements DomainEventPublisher {

    private final RocketMQTemplate rocketMQTemplate;

    @Override
    public boolean publish(String topic, String tag, String msg) {
        try {
            return this.publish(topic, tag, msg.getBytes("utf-8"));
        } catch (UnsupportedEncodingException ex) {
            ex.printStackTrace();
            throw new RuntimeException(ex);
        }
    }

    @Override
    public boolean publish(String topic, String tag, byte[] msg) {
        TimeInterval timer = DateUtil.timer();
        Message<byte[]> result = MessageBuilder.withPayload(msg)
            .setHeader(RocketMQHeaders.TAGS, tag)
            .build();
        rocketMQTemplate.syncSend(topic + COLON + tag, result);
        log.info("[producer] topic={}, tags={}", topic, tag);
//        rocketMQTemplate.syncSend(topic + COLON + tag, MessageBuilder.withPayload(msg).build());
        if (timer.interval() > 100) {
            log.warn("[producer] mq发送时间超时了。topic={}|{}", topic, tag);
        }
        return Boolean.TRUE;
    }


}
